package com.siao.net.socket;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author SiaoLeon
 * Socket连接
 */
public class Connection {

	private static final Logger logger = LoggerFactory.getLogger(Connection.class);

	/**
	 * 连接服务
	 */
	private final ConnectionService service;

	/**
	 * 读取缓存
	 */
	private final ByteBuffer readBuffer;

	/**
	 * 写入缓存队列
	 */
	private final ConcurrentLinkedQueue<IDataPack> writeBufferQueue;

	/**
	 * Socket通道
	 */
	private AsynchronousSocketChannel channel;

	/**
	 * 连接地址
	 */
	private SocketAddress address;

	/**
	 * 是否正在关闭连接
	 */
	private boolean isClosing;

	/**
	 * 是否正在写入消息
	 */
	private boolean isWriting;

	/**
	 * 是否已连接
	 * @return
	 */
	public Boolean isConnected() {
		return (this.channel == null) ? false : this.channel.isOpen();
	}

	/**
	 * 构造函数
	 * @param service
	 */
	public Connection(ConnectionService service) {
		this.service = service;
		this.readBuffer = ByteBuffer.allocate(this.service.bufferSize);
		this.writeBufferQueue = new ConcurrentLinkedQueue<IDataPack>();
		this.isClosing = false;
		this.isWriting = false;
	}

	public String toString() {
		return (this.address == null) ? "Connection" : "Connection(" + this.address + ")";
	}

	/**
	 * 关闭连接
	 */
	public final void close() {
		if (!this.isConnected() || this.isClosing) {
			return;
		}

		this.isClosing = true;

		this.service.connectionClose(this);
		this.write();
	}

	/**
	 * 发送数据
	 * @param bytes
	 */
	public final void send(IDataPack data) {
		if (!this.isConnected() || this.isClosing) {
			return;
		}

		this.writeBufferQueue.add(data);
		this.write();
	}

	/**
	 * 将连接注册在指定Socket通道上
	 * @param channel
	 */
	final void register(AsynchronousSocketChannel channel) {
		try {
			this.channel = channel;
			this.address = (channel.isOpen()) ? channel.getRemoteAddress() : null;
			this.service.connectionRegister(this);
			this.read();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 断开连接
	 */
	final synchronized void disconnect() {
		if (!this.isConnected()) {
			return;
		}

		try {
			this.channel.shutdownInput();
			this.channel.shutdownOutput();
			this.channel.close();
		} catch (IOException e) {
			logger.error("Failed to disconnect " + this + ".");
		}

		this.service.connectionDisconnect(this);

		this.channel = null;
		this.address = null;
		this.readBuffer.clear();
		this.writeBufferQueue.clear();
		this.isClosing = false;
		this.isWriting = false;
	}

	private synchronized final void read() {
		if (!this.isConnected() || this.isClosing) {
			return;
		}

		this.channel.read(this.readBuffer, this.readBuffer, new Reader());
	}

	private synchronized final void write() {
		if (!this.isConnected() || this.isWriting) {
			return;
		}

		if (!this.writeBufferQueue.isEmpty()) {
			this.isWriting = true;
			IDataPack data = this.writeBufferQueue.poll();

			ByteBuffer buffer = (this.service.dataResolver == null) ? ByteBuffer.wrap(data.read()) : this.service.dataResolver.pack(data);
			buffer.position(0);
			this.channel.write(buffer, data, new Writter());
		}

		if (this.isClosing && !this.isWriting) {
			this.disconnect();
		}
	}

	/**
	 * @author SiaoLeon
	 * 读取处理器
	 */
	private final class Reader implements CompletionHandler<Integer, ByteBuffer> {

		final Connection connection;

		public Reader() {
			this.connection = Connection.this;
		}

		@Override
		public void completed(Integer result, ByteBuffer buffer) {
			synchronized (this.connection) {
				if (!this.connection.isConnected() || this.connection.isClosing) {
					return;
				}

				if (result < 0) {
					this.connection.disconnect();
					return;
				}

				buffer.flip();

				try {
					if (this.connection.service.dataResolver == null) {
						byte[] bytes = new byte[buffer.remaining()];
						buffer.get(bytes);
						this.connection.service.connectionReceive(this.connection, bytes);
					} else {
						while (this.connection.service.dataResolver.isContains(buffer)) {
							this.connection.service.connectionReceive(this.connection, this.connection.service.dataResolver.parse(buffer));
						}
					}

				} catch (Exception e) {
					this.connection.disconnect();
				}

				buffer.compact();
			}

			this.connection.read();
		}

		@Override
		public void failed(Throwable exc, ByteBuffer buffer) {
			if (exc instanceof IOException) {
				this.connection.disconnect();
			}
		}

	}

	/**
	 * @author SiaoLeon
	 * 写入处理器
	 */
	private final class Writter implements CompletionHandler<Integer, IDataPack> {

		final Connection connection;

		public Writter() {
			this.connection = Connection.this;
		}

		@Override
		public void completed(Integer result, IDataPack data) {
			synchronized (this.connection) {
				if (!this.connection.isConnected() || !this.connection.isWriting) {
					return;
				}

				if (result < 0) {
					System.out.println(result);
				}

				this.connection.isWriting = false;

				this.connection.service.connectionSend(this.connection, data);
			}

			this.connection.write();
		}

		@Override
		public void failed(Throwable exc, IDataPack data) {
			if (exc instanceof IOException) {
				this.connection.disconnect();
			}
		}

	}

}
